package com.shujia.state

import org.apache.flink.api.common.functions.{MapFunction, ReduceFunction, RichMapFunction, RuntimeContext}
import org.apache.flink.api.common.state.{ReducingState, ReducingStateDescriptor}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala._

object Demo05ReducingState {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val linesDS: DataStream[String] = env.socketTextStream("master", 8888)

    /**
     * 使用ReducingState统计并保存每个单词的数量
     */
    linesDS
      .flatMap(_.split(","))
      .map(word => (word, 1))
      .keyBy(_._1)
      .map(new RichMapFunction[(String, Int), (String, Int)] {
        var reducingState: ReducingState[Int] = _

        override def open(parameters: Configuration): Unit = {
          val ctx: RuntimeContext = getRuntimeContext
          val reducingStateDesc: ReducingStateDescriptor[Int] = new ReducingStateDescriptor[Int]("cnt", new ReduceFunction[Int] {
            override def reduce(value1: Int, value2: Int): Int = {
              value1 + value2
            }
          }, classOf[Int])
          reducingState = ctx.getReducingState(reducingStateDesc)
        }

        override def close(): Unit = super.close()

        override def map(value: (String, Int)): (String, Int) = {
          reducingState.add(1)
          val cnt: Int = reducingState.get()
          (value._1, cnt)
        }
      })
      .print()

    env.execute()


  }

}
